package cn.com.itg.function.process;

import cn.com.itg.pojo.domain.entity.ZTMM;
import cn.com.itg.pojo.domain.vo.MatdocSendVO;
import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.io.Serializable;
import java.util.*;

@Slf4j
public class MatdocZtmmBPFunction extends KeyedBroadcastProcessFunction<String, MatdocSendVO, List<ZTMM>, MatdocSendVO> implements Serializable {
    private List<ZTMM> ztmmList = new ArrayList<>();

    @Override
    public void processElement(MatdocSendVO matdocSend, ReadOnlyContext readOnlyContext, Collector<MatdocSendVO> collector) throws Exception {
        log.info("ztmmList is :【{}】", JSON.toJSONString(ztmmList));
        ZTMM ztmmFilter = this.ztmmList.stream()
                .filter(ztmm -> Optional.ofNullable(matdocSend.getMatnr()).orElse("").endsWith(ztmm.getMatnr()))
                .max(Comparator.comparingInt(z -> z.getMatnr().length()))
                .orElse(null);
        log.info("ztmmFilter is :【{}】", JSON.toJSONString(ztmmFilter));
        collector.collect(buildSendVOByZtmm(matdocSend, ztmmFilter));
    }

    @Override
    public void processBroadcastElement(List<ZTMM> ztmms, Context context, Collector<MatdocSendVO> collector) throws Exception {
        this.ztmmList = ztmms;
    }

    private static MatdocSendVO buildSendVOByZtmm(MatdocSendVO matdocSendVO, ZTMM ztmm) {
        if (Objects.isNull(ztmm)) {
            return matdocSendVO;
        }
        matdocSendVO
                //ZTMM表
                .setMaktx(ztmm.getMaktx())
                .setZmatnr(ztmm.getJobno())
        ;
        return matdocSendVO;
    }
}
